package operator.function;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.BiPredicate;
import io.reactivex.functions.Predicate;
import util.Printer;
import util.RxJavaHelper;

class RetryOperator {

    public static void main(String[] args) {
//        retry1();
//        retry2();
//        retry3();
        retry4();
    }

    /**
     * 当发生错误时重新订阅
     */
    static void retry1() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            int start;

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                if (start > 5) {
                    Printer.print("onComplete");
                    emitter.onComplete();
                } else {
                    emitter.onNext(start);
                    Printer.print("onNext start = " + start);
                    start++;
                    emitter.onError(RxJavaHelper.getException());
                }
            }
        })
                .retry()
                .subscribe(RxJavaHelper.getConsumer());

    }

    /**
     * 当发生错误时按次数重新订阅
     */
    static void retry2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            int start;

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                if (start > 5) {
                    Printer.print("onComplete");
                    emitter.onComplete();
                } else {
                    emitter.onNext(start);
                    Printer.print("onNext start = " + start);
                    start++;
                    emitter.onError(RxJavaHelper.getException());
                }
            }
        })
                .retry(2)
                .subscribe(RxJavaHelper.getConsumer());

    }

    /**
     * 返回发生的错误信息
     */
    static void retry3() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            int start;

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                if (start < 2) {
                    Printer.print("retry when error IllegalArgumentException");
                    emitter.onError(new IllegalArgumentException());
                } else {
                    emitter.onNext(start);
                    Printer.print("onNext start = " + start);
                    emitter.onComplete();
                }
                start++;
            }
        })
                .retry(new Predicate<Throwable>() {
                    @Override
                    public boolean test(Throwable throwable) throws Exception {
                        return throwable instanceof IllegalArgumentException;
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());

    }

    /**
     * 导致发生错误的throwable以及以及尝试的次数
     * 只有错误才会触发BiPredicate的test方法
     */
    static void retry4() {
        Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("emitter onNext");
                emitter.onError(new IllegalArgumentException());
            }
        })
                .retry(new BiPredicate<Integer, Throwable>() {
                    @Override
                    public boolean test(Integer integer, Throwable throwable) throws Exception {
                        Printer.print("BiPredicate integer = " + integer);
                        Printer.print("BiPredicate throwable = " + throwable);
                        if(integer > 5){
                            return false;
                        }
                        return true;
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());

    }

}
